from datetime import timedelta
from airflow import DAG
from airflow.sdk.definitions.deadline import AsyncCallback, DeadlineAlert, DeadlineReference
from airflow.providers.discord.notifications.discord import DiscordNotifier
from airflow.providers.discord.notifications.embed import Embed, EmbedAuthor, EmbedFooter, EmbedField
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.sdk import task

with DAG(
    dag_id="deadline_discord_alert_example",
    deadline=DeadlineAlert(
        reference=DeadlineReference.DAGRUN_QUEUED_AT,
        interval=timedelta(seconds=20),
        callback=AsyncCallback(
            DiscordNotifier,
            kwargs={
                "text": "You can~~not~~ do `this`.```py\nAnd this.\nprint('Hi')```\n*italics* or _italics_     __*underline italics*__\n**bold**     __**underline bold**__\n***bold italics***  __***underline bold italics***__\n__underline__     ~~Strikethrough~~",
                "embed": Embed(
                    title="Hello ~~people~~ world :wave:",
                    description="You can use [links](https://discord.com) or emojis :smile: 😎\n```\nAnd also code blocks\n```",
                    color=4321431,
                    timestamp="2025-11-23T19:05:15.292Z",
                    url="https://discord.com",
                    author=EmbedAuthor(
                        name="Author",
                        url="https://discord.com",
                        icon_url="https://cdn.discordapp.com/embed/avatars/0.png",

                    ),
                    footer=EmbedFooter(
                        text="Footer text",
                        icon_url="https://cdn.discordapp.com/embed/avatars/0.png"
                    ),
                    fields=[
                        EmbedField(
                            name="Field 1, *lorem* **ipsum**, ~~dolor~~",
                            value="Field value"
                        )
                    ]
                )
            },
        ),
    ),
):
    c = EmptyOperator(task_id="example_task")

    @task()
    def wait():
        import time
        time.sleep(60*5)

    
    c >> wait()